package com.unitd.scheduler.monitor;

import com.alibaba.fastjson.JSON;
import com.unitd.scheduler.model.JobConfig;
import com.unitd.scheduler.model.JobGroupInfo;
import com.unitd.scheduler.registry.impl.ZkJobRegistry;
import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.ZkConnection;
import com.unitd.frame.comm.utils.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Closeable;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 * @desc Job任务的调度器监控类,用于监听定时器服务与zookeeper是否连接上,并将job注册到zookeeper
 * @filename SchedulerMonitor.java
 * @copyright www.unitd.com
 * @author Hudan
 * @version 1.0
 * @date 2016/10/14
 */
public class SchedulerMonitor implements Closeable {

	private static final Logger logger = LoggerFactory.getLogger(SchedulerMonitor.class);

	private ZkClient zkClient;

	/**
	 * @desc 构造方法,根据注册类型,实例化zookeeper客户端对象或者Redis客户端对象(目前默认zookeeper)
	 * @param registryType 注册类型
	 * @param servers Redis/Zookeeper服务器地址信息
	 */
	public SchedulerMonitor(String registryType, String servers) {

		if ("redis".equals(registryType)) {

		} else {
			ZkConnection zkConnection = new ZkConnection(servers);
			zkClient = new ZkClient(zkConnection, 3000);
		}
	}

	/**
	 * @desc 关闭当前应用与Redis/Zookeeper的连接
	 */
	@Override
	public void close() {
		if (zkClient != null) zkClient.close();
	}


	/**
	 * @desc 获取所有已注册的定时器Job任务
	 * @return 已注册的定时器组列表
	 */
	public List<JobGroupInfo> getAllJobGroups() {
		// zk registry
		List<JobGroupInfo> result = new ArrayList<>();

		// 获取所有已向zookeeper注册的定时任务组
		List<String> groupNames = getGroups();
		if (groupNames == null) return result;
		for (String groupName : groupNames) {
			JobGroupInfo groupInfo = getJobGroupInfo(groupName);

			if (groupInfo != null) {
				result.add(groupInfo);
			}
		}
		return result;
	}

	/**
	 * @desc 根据定时任务组名获取该组下的已注册定时器任务信息
	 * @param groupName 定时任务组名
	 * @return JobGroupInfo
	 */
	public JobGroupInfo getJobGroupInfo(String groupName) {

		if (StringUtils.isBlank(groupName)) {
			logger.warn("getJobGroupInfo groupName is required");
			return null;
		}
		JobGroupInfo groupInfo = new JobGroupInfo();
		groupInfo.setName(groupName);

		String path = ZkJobRegistry.ROOT + groupName;
		List<String> children = zkClient.getChildren(path);
		for (String child : children) {
			if ("nodes".equals(child)) {
				path = ZkJobRegistry.ROOT + groupName + "/nodes";
				groupInfo.setClusterNodes(zkClient.getChildren(path));
			} else {
				path = ZkJobRegistry.ROOT + groupName + "/" + child;
				Object data = zkClient.readData(path);
				if (data != null) {
					JobConfig jobConfig = JSON.parseObject(data.toString(), JobConfig.class);
					groupInfo.getJobs().add(jobConfig);
				}
			}
		}

		if (groupInfo.getClusterNodes().size() > 0) {
			return groupInfo;
		}
		return null;
	}

	/**
	 * @desc 获取所有已向zookeeper注册的定时任务组信息
	 * @return List
	 */
	public List<String> getGroups() {
		String path = ZkJobRegistry.ROOT.substring(0, ZkJobRegistry.ROOT.length() - 1);
		return zkClient.getChildren(path);
	}

	/**
	 * @desc 根据命令信息,响应命令,执行响应的定时任务
	 * @param cmd 定时任务命令信息
	 */
	public void publishEvent(MonitorCommond cmd) {
		String path = ZkJobRegistry.ROOT + cmd.getJobGroup() + "/nodes";
		List<String> nodeIds = zkClient.getChildren(path);

		for (String node : nodeIds) {
			String nodePath = path + "/" + node;
			zkClient.writeData(nodePath, cmd);
			logger.info("publishEvent finish，path:{},content:{}", nodePath, cmd);
			break;
		}
	}

	/**
	 * @desc 清理已失效的定时任务组(包含该任务组下的定时任务一并清除掉)
	 */
	public void clearInvalidGroup() {
		List<String> groups = zkClient.getChildren(ZkJobRegistry.ROOT.substring(0, ZkJobRegistry.ROOT.length() - 1));
		logger.info("==============clear Invalid jobs=================");
		for (String group : groups) {
			String groupPath = ZkJobRegistry.ROOT + group;
			String nodeStateParentPath = groupPath + "/nodes";
			try {
				if (!zkClient.exists(nodeStateParentPath)
						|| zkClient.countChildren(nodeStateParentPath) == 0) {
					List<String> jobs = zkClient.getChildren(groupPath);
					for (String job : jobs) {
						zkClient.delete(groupPath + "/" + job);
						logger.info("delete path:{}/{}", groupPath, job);
					}
					zkClient.delete(groupPath);
					logger.info("delete path:{}", groupPath);
				}
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
		logger.info("==============clear Invalid jobs end=================");
	}

	public static void main(String[] args) throws IOException {
		SchedulerMonitor monitor = new SchedulerMonitor("zookeeper", "127.0.0.1:2181");

		List<JobGroupInfo> groups = monitor.getAllJobGroups();
		System.out.println(JSON.toJSONString(groups));

		monitor.close();
	}
}